package com.seari.service.impl;

import com.seari.bean.DataTable;
import com.seari.bean.DataTableKey;
import com.seari.bean.RedisKey;
import com.seari.common.Const;
import com.seari.dao.FirstInitDaoImpl;
import com.seari.service.CommonDataService;
import com.seari.service.FirstInitService;
import com.seari.utils.*;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;
import redis.clients.jedis.Jedis;
import redis.clients.jedis.Pipeline;

import java.util.*;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.atomic.AtomicInteger;

@Service("firstInitService")
public class FirstInitServiceImpl implements FirstInitService{

    private static Logger logger = LoggerFactory.getLogger(FirstInitServiceImpl.class);

    public static final int MAX_WORKER_COUNT = 10;
    public static final int MAX_SLICE_COUNT = 20000;
    public static BlockingQueue<List<String>> aiQueue = new LinkedBlockingQueue<>(200);
    public static BlockingQueue<List<String>> diQueue = new LinkedBlockingQueue<>(200);
    public static ExecutorService service = Executors.newCachedThreadPool();
    public static AtomicInteger workerCount = new AtomicInteger();

    Jedis jedis = JedisPoolUtils.getJedis();

    GuavaLocalCacheUtils guavaLocalCacheUtils = new GuavaLocalCacheUtils();

    List<RedisKey> aiKeyList = new ArrayList<RedisKey>();
    List<RedisKey> diKeyList = new ArrayList<RedisKey>();

    @Autowired
    private CommonDataService commonDataService;

    @Override
    public void initLocalCacheAndPutDataToInfluxdb() {

        initRedisDataToLocalCache();
        logger.info("init initRedisDataToLocalCache complete");
        initMysqlDataToLocalCache();
        logger.info("init initMysqlDataToLocalCache complete");

        if (!CommonUtils.objectIsNull(aiKeyList)) {
            for (int i = 0;i<aiKeyList.size();i++) {
                String aikeyStr = aiKeyList.get(i).getKeyNameEn();
                Set<String> aiKeyset = jedis.hkeys(aikeyStr);
                if(!CommonUtils.objectIsNull(aiKeyset)) {
                    List<String> result = new ArrayList<String>(aiKeyset);
                    //AI数据 有的值需要和下一位的合并   combine 合并标志（为1时合并后一个redis地址中的值 ，仅用于AI中的FLOAT量）  这里不做分割集合了
                    List<String> listStr = new ArrayList<String>();
                    listStr.addAll(result);
                    listStr.add(aikeyStr);
                    aiQueue.add(listStr);
                }else{
                    logger.info(" init  "+aikeyStr+ " key  not find in redis");
                }
            }
        }else{
            logger.info(" init  " + aiKeyList + " aiKeyList is null");
        }

        if (!CommonUtils.objectIsNull(diKeyList)) {
            for (int j = 0;j<aiKeyList.size();j++) {
                String dikeyStr = diKeyList.get(j).getKeyNameEn();
                Set<String> diKeyset = jedis.hkeys(dikeyStr);
                if(!CommonUtils.objectIsNull(diKeyset)) {
                    List<String> result = new ArrayList<String>(diKeyset);
                    List<List<String>> listList = ListUtils.splitList(result, MAX_SLICE_COUNT);
                    for (List<String> list : listList) {
                        List<String> listStr = new ArrayList<String>();
                        listStr.addAll(list);
                        listStr.add(dikeyStr);
                        diQueue.add(listStr);
                    }
                }else{
                    logger.info(" init  "+dikeyStr+ "  key not find in redis");
                }
            }
        }else{
            logger.info(" init  " + diKeyList + " diKeyList is null");
        }

        startWorker();
    }

    @Override
    public void initRedisDataToLocalCache(){

        aiKeyList = commonDataService.getRedisKeysByType(Const.REDIS_AI_KEY_TYPE);
        diKeyList = commonDataService.getRedisKeysByType(Const.REDIS_DI_KEY_TYPE);

        if (!CommonUtils.objectIsNull(aiKeyList)) {
            Pipeline pipeline = jedis.pipelined();
            for (int i = 0; i < aiKeyList.size(); i++) {
                pipeline.hgetAll(aiKeyList.get(i).getKeyNameEn());
            }
            List<Object> aiResults = pipeline.syncAndReturnAll();
            for(int i = 0; i < aiKeyList.size(); i++){
                String keyStr = aiKeyList.get(i).getKeyNameEn();
                Object valueStr = aiResults.get(i);
                guavaLocalCacheUtils.cache.put(keyStr, valueStr);
            }
        }

        if (!CommonUtils.objectIsNull(diKeyList)) {
            Pipeline pipeline = jedis.pipelined();
            for (int j = 0; j < diKeyList.size(); j++) {
                pipeline.hgetAll(diKeyList.get(j).getKeyNameEn());
            }
            List<Object> diResults = pipeline.syncAndReturnAll();
            for(int j = 0; j < diKeyList.size(); j++){
                String keyStr = diKeyList.get(j).getKeyNameEn();
                Object valueStr = diResults.get(j);
                guavaLocalCacheUtils.cache.put(keyStr, valueStr);
            }
        }
    }

    /**
     *
     */
    @Override
    public void initMysqlDataToLocalCache(){

        for(RedisKey redisKey : aiKeyList) {
            List<DataTable> dataTableList = commonDataService.getDataTableListByKey(redisKey.getKeyNameEn());
            logger.info("FirstInitServiceImpl.initMysqlDataToLocalCache.dataTableList.aiKeyList.size=" + dataTableList.size());
            List<DataTableKey> dataTableKeyList = commonDataService.getDataTableKeyList(redisKey.getKeyNameEn());
            logger.info("FirstInitServiceImpl.initMysqlDataToLocalCache.dataTableKeyList.aiKeyList.size="+dataTableKeyList.size());

            if (!CommonUtils.objectIsNull(dataTableList)) {
                for(DataTableKey dataTableKey : dataTableKeyList) {
                    List<DataTable> dataTableRedisKey = new ArrayList<DataTable>();
                    for (DataTable dataTable : dataTableList) {
                        if (dataTableKey.getRedisAddress().equals(dataTable.getRedisAddress())) {
                            dataTableRedisKey.add(dataTable);
                        }
                    }
                    guavaLocalCacheUtils.cache.put(dataTableKey.getTableName()+dataTableKey.getRedisAddress(), dataTableRedisKey);
                 }
             }
          }


        for(RedisKey redisKey : diKeyList) {
            List<DataTable> dataTableList = commonDataService.getDataTableListByKey(redisKey.getKeyNameEn());
            logger.info("FirstInitServiceImpl.initMysqlDataToLocalCache.dataTableList.diKeyList.size=" + dataTableList.size());
            List<DataTableKey> dataTableKeyList = commonDataService.getDataTableKeyList(redisKey.getKeyNameEn());
            logger.info("FirstInitServiceImpl.initMysqlDataToLocalCache.dataTableKeyList.diKeyList.size="+dataTableKeyList.size());

            if (!CommonUtils.objectIsNull(dataTableList)) {
                for(DataTableKey dataTableKey : dataTableKeyList) {
                    List<DataTable> dataTableRedisKey = new ArrayList<DataTable>();
                    for (DataTable dataTable : dataTableList) {
                        if (dataTableKey.getRedisAddress().equals(dataTable.getRedisAddress())) {
                            dataTableRedisKey.add(dataTable);
                        }
                    }
                    guavaLocalCacheUtils.cache.put(dataTableKey.getTableName()+dataTableKey.getRedisAddress(), dataTableRedisKey);
                }
            }
        }
    }

    @Override
    public void startWorker(){
        while(workerCount.get() < MAX_WORKER_COUNT){
            FirstInitDaoImpl FirstInitDaoWorker = new FirstInitDaoImpl(aiQueue,diQueue);
            service.submit(FirstInitDaoWorker);
            int num = workerCount.incrementAndGet();
            logger.info("start worker:current work count is:" + num);
        }
    }
}
